package com.atguigu.ting.realtime.app.dwd.db;

import com.atguigu.ting.realtime.app.BaseSqlApp;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/4/11 10:08
 */
public class DwdTradePaySuccess extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradePaySuccess().init(
            3003,
            2,
            "DwdTradePaySuccess"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        // 1. 读取 ods_db
        readOdsDb(tEnv, "DwdTradePaySuccess");
        // 2. 读取字典表
        readBaseDic(tEnv);
        // 3. 读取下单事务事实表
        tEnv.executeSql(
            "create table dwd_trade_order_detail(" +
                "order_id string, " +
                "user_id string, " +
                "order_detail_id string, " +
                "trade_no string, " +
                "item_id string, " +
                "item_name string, " +
                "item_type string, " +
                "item_type_name string, " +
                "derate_type string, " +
                "derate_type_name string, " +
                "split_original_amount decimal(16,2), " +
                "split_derate_amount decimal(16,2), " +
                "split_order_amount decimal(16,2), " +
                "create_time string, " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "DwdTradePaySuccess1"));
        // 4. 过滤支付成功表
        Table paymentInfo = tEnv.sqlQuery(
            "select " +
                "`data`['id'] id, " +
                "`data`['trade_no'] trade_no, " +
                "`data`['payment_type'] payment_type, " +
                "`data`['update_time'] payment_time, " +
                "pt, " +
                "ts " +
                "from ods_db " +
                "where `database`='ting' " +
                "and `table`='payment_info' " +
                "and `type`='update' " +
                "and `old`['payment_status']='1601' " +
                "and `data`['payment_status']='1602' ");
        tEnv.createTemporaryView("payment_info", paymentInfo);
        
        // 5. 3 张表 join
        Table result = tEnv.sqlQuery(
            "select " +
                "pi.id id, " +
                "od.order_id order_id, " +
                "od.user_id user_id, " +
                "od.order_detail_id order_detail_id, " +
                "od.item_id item_id, " +
                "od.item_name item_name, " +
                "od.item_type item_type, " +
                "od.item_type_name item_type_name, " +
                "od.derate_type derate_type, " +
                "od.derate_type_name derate_type_name, " +
                "pi.trade_no trade_no, " +
                "pi.payment_type payment_type," +
                "dic.dic_name payment_type_name, " +
                "od.split_original_amount split_original_amount, " +
                "od.split_derate_amount split_derate_amount, " +
                "od.split_order_amount split_order_amount, " +
                "pi.payment_time payment_time, " +
                "pi.ts ts " +
                "from payment_info pi " +
                "join dwd_trade_order_detail od " +
                "on pi.trade_no=od.trade_no " +
                "join base_dic for system_time as of pi.pt as dic " +
                "on pi.payment_type=dic.dic_code ");
        
        // 6. 写出到 kafka 中
        tEnv.executeSql(
            "create table dwd_trade_pay_success(" +
                "id string, " +
                "order_id string, " +
                "user_id string, " +
                "order_detail_id string, " +
                "item_id string, " +
                "item_name string, " +
                "item_type string, " +
                "item_type_name string, " +
                "derate_type string, " +
                "derate_type_name string, " +
                "trade_no string, " +
                "payment_type string," +
                "payment_type_name string, " +
                "split_original_amount decimal(16, 2), " +
                "split_derate_amount decimal(16, 2), " +
                "split_order_amount decimal(16, 2), " +
                "payment_time string, " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_PAY_SUCCESS));
    
        result.executeInsert("dwd_trade_pay_success");
        
    }
}
